package dnet

import (
	"dgo/goutils/utils"
	"fmt"
	"net"
	"sync"
	"sync/atomic"
	"time"
)

type EventBuffer func(client *ASyncTcpClient, buffer []byte)
type EventTcpClient func(client *ASyncTcpClient)
type EventTcpClientResult func(client *ASyncTcpClient) interface{}

/*
**

	推荐使用ASyncTcpClientEx
*/
type ASyncTcpClient struct {
	sendBytes int64

	object_id   uint32
	startflag   int32
	start_t     time.Time
	stop_t      time.Time
	workingflag int32

	IsTerminated bool
	Connected    bool

	Conn              net.Conn
	RemoteAddress     string
	RecvBlockSize     uint16
	sendQueueSize     int
	sendQueue         chan interface{}
	OnConnected       EventTcpClient
	OnDisconnected    EventTcpClient
	OnConnectFail     EventTcpClient
	OnRecvBuffer      EventBuffer
	LastConnectErr    error
	UseQueueSend      bool // 使用队列进行发送
	LastDisMsg        string
	closeWg           *sync.WaitGroup
	lastSendTime      time.Time
	disTime           time.Time     // 断线时间
	reconnectInterval time.Duration // 重连间隔
	LastRecvTime      time.Time     // 最后一次接收数据的时间
	BeforeConnectCb   EventTcpClientResult
	DailCb            func(network, addr string) (c net.Conn, err error)
}

func NewASyncTcpClient() *ASyncTcpClient {
	client := &ASyncTcpClient{
		object_id:         utils.NewObjectDNA(),
		IsTerminated:      false,
		Connected:         false,
		startflag:         0,
		reconnectInterval: time.Second * 5,
		RecvBlockSize:     RECV_BLOCK_SIZE,
		UseQueueSend:      true,
		closeWg:           new(sync.WaitGroup),
	}
	client.sendQueueSize = 1024
	client.sendQueue = make(chan interface{}, client.sendQueueSize)
	return client
}

func (this *ASyncTcpClient) GetObjectID() uint32 {
	return this.object_id
}

func (this *ASyncTcpClient) SendBytes() int64 {
	return this.sendBytes
}

func (this *ASyncTcpClient) SetAddress(addr string) {
	this.RemoteAddress = addr
}

func (this *ASyncTcpClient) IsActive() bool {
	return this.startflag == 1
}

/*
start 和Stop 最好在同一个纤程中运行. 不要进行并发
*/
func (this *ASyncTcpClient) Start() {
	if atomic.CompareAndSwapInt32(&this.startflag, 0, 1) {
		this.start_t = time.Now()
		this.Connected = false
		this.IsTerminated = false
		if this.UseQueueSend {
			utils.GoFunc2(this.handleSend)
		}
		utils.GoFunc2(this.HandRecv)
		time.Sleep(time.Millisecond) // 等待开启
	} else {
		utils.Log.Warn("连接已经打开！请勿重复启动")
	}
}

func (this *ASyncTcpClient) Stop() {
	this.IsTerminated = true
	this.innerCheckClose()
}

func (this *ASyncTcpClient) BreakConnect() {
	if this.Connected {
		this.Conn.Close()
	}
}

func (this *ASyncTcpClient) SendBufferDirect(buffer []byte) {
	defer utils.PanicHandlerWithPrint()
	if !this.Connected {
		utils.Log.Warnf("尚未成功建立连接(%s)，数据包丢弃!", this.RemoteAddress)
	}

	if this.Conn == nil {
		utils.Log.Infof("连接(%s)不存在,或者尚未建立,数据包丢弃!", this.RemoteAddress)
	}

	n, err := this.Conn.Write(buffer)
	if err != nil {
		utils.Log.Infof("%s 处理发送数据时出现了异常:%v", this.Conn.RemoteAddr().String(), err)
	} else {
		atomic.AddInt64(&this.sendBytes, int64(n))
		this.lastSendTime = time.Now()
	}

}

func (this *ASyncTcpClient) PostSendBuffer(buffer []byte) int {
	if this.Connected {
		if this.UseQueueSend {
			if len(this.sendQueue) < this.sendQueueSize {
				this.sendQueue <- buffer
				return len(buffer)
			} else {
				utils.Log.Infof("%s 处理发送数据时出现了异常:发送队列堆积[%d]", this.Conn.RemoteAddr().String(), this.sendQueueSize)
				return 0
			}
		} else {
			n, err := this.Conn.Write(buffer)
			if err != nil {
				utils.Log.Infof("%s 处理发送数据时出现了异常:%v", this.Conn.RemoteAddr().String(), err)
				this.BreakConnect()
			}
			atomic.AddInt64(&this.sendBytes, int64(n))
			this.lastSendTime = time.Now()

			return n
		}

	} else {
		utils.Log.Debugf("连接(%s)已经断开，不投递到发送队列！", this.RemoteAddress)
		return 0
	}
}

func (this *ASyncTcpClient) innerCheckDoConnect() bool {
	this.LastConnectErr = nil
	this.LastDisMsg = ""

	if len(this.RemoteAddress) == 0 {
		return false
	}

	if this.BeforeConnectCb != nil {
		r := this.BeforeConnectCb(this)
		if r.(bool) {

		} else {
			return false
		}
	}
	var conn net.Conn
	var err error
	if this.DailCb != nil {
		conn, err = this.DailCb("tcp", this.RemoteAddress)
	} else {
		conn, err = net.Dial("tcp", this.RemoteAddress)
	}

	if err != nil {
		utils.Log.Debugf(" [%d][%s]连接出现异常：%v", this.object_id, this.RemoteAddress, err)
		this.LastConnectErr = err
		if this.OnConnectFail != nil {
			this.OnConnectFail(this)
		}
		return false
	}
	this.Conn = conn
	this.Connected = true
	if this.OnConnected != nil {
		this.OnConnected(this)
	}
	this.LastRecvTime = time.Now()

	return true
}

// 接收数据纤程调用
func (this *ASyncTcpClient) innerDoClose() {
	if this.Conn != nil {
		this.Conn.Close()
	}

	if this.Connected {
		if atomic.CompareAndSwapInt32(&this.workingflag, 0, 1) { // 避免多次触发
			if this.OnDisconnected != nil {
				this.OnDisconnected(this)
			}

			// 等待连接
			this.Connected = false
			this.disTime = time.Now()
			this.workingflag = 0
		}
	}

}

func (this *ASyncTcpClient) innerCheckClose() {
	if atomic.CompareAndSwapInt32(&this.startflag, 1, 0) {
		this.stop_t = time.Now()
		this.IsTerminated = true
		if this.UseQueueSend {
			// 请求退出发送队列
			this.sendQueue <- nil
		}
		this.LastDisMsg = "主动关闭连接"
		utils.Log.Debugf("[%d][%s]主动关闭连接", this.object_id, this.RemoteAddress)
		this.innerDoClose()
		this.closeWg.Wait()
	} else {
		utils.Log.Debugf("[%d]连接尚未开启", this.object_id)
	}
}

func (this *ASyncTcpClient) handleSend() {
	this.closeWg.Add(1)
	defer this.closeWg.Done()

sendfor:
	for {
		if !this.Connected {
			if this.IsTerminated {
				break sendfor
			}
			// 等待重连
			time.Sleep(time.Second)
			continue
		}

		select {
		case data, ok := <-this.sendQueue:
			if ok && data != nil {
				if this.Conn == nil {
					continue
				}
				_, err := this.Conn.Write(data.([]byte))
				if err != nil {
					if this.Conn == nil {
						continue
					}
					utils.Log.Debugf("[%d][%s]处理发送数据时出现了异常:%s", this.object_id, this.RemoteAddress, err.Error())
					this.BreakConnect()
				}
			} else {
				break sendfor
			}
		}
	}

	utils.Log.Debugf("[%d][%s]循环发送任务工作退出！", this.object_id, this.RemoteAddress)

}

// / 如果超时没有收到数据(返回true
func (this *ASyncTcpClient) CheckSendTimeOut(isec int) bool {
	if !this.Connected {
		return false
	}

	if this.lastSendTime.IsZero() {
		return false
	}

	t := time.Since(this.lastSendTime)
	tick := int(t.Seconds())
	if tick > isec {
		return true
	} else {
		return false
	}
}

// / 如果超时没有收到数据(返回true
func (this *ASyncTcpClient) CheckRecvTimeOut(isec int) bool {
	if !this.Connected {
		return false
	}

	if this.LastRecvTime.IsZero() {
		return false
	}

	t := time.Since(this.LastRecvTime)
	tick := int(t.Seconds())
	if tick > isec {
		return true
	} else {
		return false
	}
}

// / 如果返回true, 表示断开了一次连接
func (this *ASyncTcpClient) CheckBreakConnectIfTimeOut(isec int) bool {
	recvTimeOut := this.CheckRecvTimeOut(isec)
	sendTimeOut := this.CheckSendTimeOut(isec)
	if recvTimeOut && sendTimeOut {
		this.BreakConnect()
		return true
	} else {
		return false
	}
}

// / 如果返回true, 表示断开了一次连接
func (this *ASyncTcpClient) CheckBreakConnectIfRecvTimeOut(isec int) bool {
	if this.CheckRecvTimeOut(isec) {
		this.BreakConnect()
		return true
	} else {
		return false
	}
}

func (this *ASyncTcpClient) HandRecv() {
	this.closeWg.Add(1)
	defer this.closeWg.Done()

	_connectflag := true
	var buffer []byte = nil
recvfor:
	for {
		if !this.Connected {
			if this.IsTerminated {
				break recvfor
			}

			if this.disTime.IsZero() {
				_connectflag = true
			} else if time.Since(this.disTime) >= this.reconnectInterval {
				_connectflag = true
			} else {
				// 重连时间没到继续等待
				time.Sleep(time.Second)
				continue
			}

			if _connectflag {
				if this.IsTerminated {
					break recvfor
				}

				if this.innerCheckDoConnect() {
					continue
				} else {
					// 重连失败
					time.Sleep(time.Second)
					continue
				}
			}
		}

		// 避免异步修改
		if RECV_MAKE_NEW_BUFF || buffer == nil {
			buffer = make([]byte, this.RecvBlockSize)
		}

		if this.IsTerminated {
			this.innerDoClose()
			break recvfor
		}

		n, err := this.Conn.Read(buffer)
		if err != nil {
			if len(this.LastDisMsg) == 0 {
				this.LastDisMsg = fmt.Sprintf(" 接受数据时出现了异常: %s", err.Error())
			}
			utils.Log.Debugf("[%d][%s]处理接受时出现了异常:%s", this.object_id, this.RemoteAddress, err.Error())

			this.innerDoClose()

			if this.IsTerminated {
				break recvfor
			}
		} else {
			if this.IsTerminated {
				this.innerDoClose()
				break recvfor
			}
			this.LastRecvTime = time.Now()
			if this.OnRecvBuffer != nil {
				this.OnRecvBuffer(this, buffer[:n])
			}
		}

	}

	utils.Log.Debugf("[%d][%s]循环接收任务工作退出！", this.object_id, this.RemoteAddress)
}
